package com.cyy.demo.common;

import com.cyy.demo.entity.User;
import com.cyy.demo.utils.Object2ArrayUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
public class KafkaConsumer {

    @KafkaListener(topics = {"test"}, containerFactory = "kafkaListenerContainerFactory")
    public void listen(List<ConsumerRecord<byte[], byte[]>> records, Acknowledgment ack) {
        boolean flag = true;
        log.info("==== start consum kafka data ====");
        if (records == null || records.isEmpty()) {
            log.warn("==== kafka data is empty =====");
        }
        try {
            for (ConsumerRecord<byte[], byte[]> record : records) {
                Object object = Object2ArrayUtil.byteArrayToObject(record.value());
                User user = (User)object;
                System.out.println(user);
            }
        } catch (Exception e) {
            log.error("consumer  data exception : {}", e);
            flag = false;
        } finally {
            if (flag) {
                ack.acknowledge();// 手动提交偏移量
            }
        }
    }


}
